package SinkTest

import SourceTest.Source01.SensorReading
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}

object RedisSink {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val stream1 = env.readTextFile("path")

    val config = new FlinkJedisPoolConfig.Builder().setHost("localhost").setPort(6379).build()
//    stream1.addSink(new RedisSink[SensorReading](config, new MyRedisMapper))


    env.execute()
  }

  class MyRedisMapper extends RedisMapper[SensorReading]{
    override def getCommandDescription: RedisCommandDescription = {
      new RedisCommandDescription(RedisCommand.HSET, "sensor_temperature")
    }

    override def getKeyFromData(t: SensorReading): String = {
      t.temperature.toString
    }

    override def getValueFromData(t: SensorReading): String = t.id
  }


}
